package com.lzy.netty.websocket;

import com.a3test.component.idworker.IdWorkerConfigBean;
import com.a3test.component.idworker.Snowflake;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.lzy.enums.MsgTypeEnum;
import com.lzy.grace.result.GraceJSONResult;
import com.lzy.netty.mq.MessagePublisher;
import com.lzy.netty.utils.JedisPoolUtils;
import com.lzy.netty.utils.ZookeeperRegister;
import com.lzy.pojo.netty.ChatMsg;
import com.lzy.pojo.netty.DataContent;
import com.lzy.pojo.netty.NettyServerNode;
import com.lzy.utils.JsonUtils;
import com.lzy.utils.LocalDateUtils;
import com.lzy.utils.OkHttpUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import redis.clients.jedis.Jedis;

import java.time.LocalDateTime;
import java.util.List;

/**
 * @author lzy
 * @date 2025/8/26 16:05
 * @description http处理类
 **/

// SimpleChannelInboundHandler<HttpObject> 泛型参数表示处理什么类型的数据 相当于入站
// TextWebSocketFrame: 用于对未WebSocket的请求处理的文本数据对象，Frame是数据（消息）的载体
public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    // 用于记录和管理所有客户端对应的channel的集合
    public static ChannelGroup clients =
                    new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext,
                                TextWebSocketFrame textWebSocketFrame) throws Exception {

        // 获得客户端传输过来的消息
        String content = textWebSocketFrame.text();
        System.out.println("服务端收到消息：" + content);

        // 1. 获取客户端发来的信息，并进行解析
        DataContent dataContent = JsonUtils.jsonToPojo(content, DataContent.class);
        ChatMsg chatMsg = dataContent.getChatMsg();
        String msgText = chatMsg.getMsg();
        String receiverId = chatMsg.getReceiverId();
        String senderId = chatMsg.getSenderId();

        // 判断是否是黑名单，start
        // 如果双方只要有一方是黑名单，则终止发送
        // friendId1st,friendId2nd
        GraceJSONResult result = OkHttpUtil.get("http://127.0.0.1:1000/friendship/isBlack?friendId1st=" + receiverId
                + "&friendId2nd=" + senderId);
        boolean isBlack = (Boolean)result.getData();
        System.out.println("当前的黑名单关系为:" + isBlack);
        if(isBlack){
            return ;
        }
        // 判断是否是黑名单，end


        // 时间校准，以服务器的时间为准
        chatMsg.setChatTime(LocalDateTime.now());

        Integer msgType = chatMsg.getMsgType();
        // 获取Channel
        Channel channel = channelHandlerContext.channel();
        String currentChannleId = channel.id().asLongText();

        // 2. 判断消息类型，根据不同的消息类型来处理不同的业务
        if(msgType == MsgTypeEnum.CONNECT_INIT.type) {
            // 当Websocket首次连接时，初始化channel，把channel和userid关联起来
            UserChannelSession.bindMultiSession(senderId, channel);
            UserChannelSession.bindUser(currentChannleId, senderId);

            // 获得当前netty集群中在线人数最少的node结点 人数 + 1
            NettyServerNode minNode = dataContent.getServerNode();
            ZookeeperRegister.incrementOnlineCounts(minNode);

            Jedis jedis = JedisPoolUtils.getJedis();
            jedis.set(senderId,JsonUtils.objectToJson(minNode));

        }
        else if(msgType == MsgTypeEnum.WORDS.type
             || msgType == MsgTypeEnum.IMAGE.type
             || msgType == MsgTypeEnum.VIDEO.type
             || msgType == MsgTypeEnum.VOICE.type
        ){
            // 此处为mq异步解耦，保存信息到数据库，数据库无法获得信息的主键id，
            // 所以此处可以使用snowflake生成id
            Snowflake snowflake = new Snowflake(new IdWorkerConfigBean());
            String sid = snowflake.nextId();
            System.out.println("sid = " + sid);
            chatMsg.setMsgId(sid);

            // 发送信息
            List<Channel> receiveChannels = UserChannelSession.getMultiChannels(receiverId);
            if(receiveChannels == null || receiveChannels.size() == 0){
                // channel为空，表示接收方为离线或者断线状态,消息不需要发送，后续可以存储到数据库中
                chatMsg.setIsReceiverOnLine(false);
            }else{
                chatMsg.setIsReceiverOnLine(true);
                // 当receiveChannels不为空，则说明接收方在线，此时发送消息给接收方所有的在线设备
                sendMsg(receiveChannels, dataContent);
            }
            // 把聊天信息作为mq的消息发送给消费者，进行消费处理 -- 保存到数据库
            MessagePublisher.sendMsgToSave(chatMsg);
        }
        // 获取当前用户的其它设备
        List<Channel> myOtherChannels = UserChannelSession.
                getMyOtherChannels(senderId,currentChannleId);
        // 同步消息给发送方在线的其他设备
        sendMsg(myOtherChannels, dataContent);
        UserChannelSession.outputMulti();
    }

    private void sendMsg(List<Channel> channels, DataContent dataContent) {

        if(channels == null || channels.size() == 0){
            return;
        }
        ChatMsg chatMsg = dataContent.getChatMsg();
        for (Channel c : channels) {
            Channel findChannel = clients.find(c.id());   //判断Channel是否在ChannelGroup中，是否有效
            if(findChannel != null){
                // 语音消息刚开始发送的时候，设置未读
                if(chatMsg.getMsgType() == MsgTypeEnum.VOICE.type){
                    chatMsg.setIsRead(false);
                }
                dataContent.setChatMsg(chatMsg);
                String chatTimeFormat = LocalDateUtils
                        .format(chatMsg.getChatTime(), LocalDateUtils.DATETIME_PATTERN_2);
                dataContent.setChatTime(chatTimeFormat);
                // 同步消息给在线的其他设备
                findChannel.writeAndFlush(
                        new TextWebSocketFrame(
                                JsonUtils.objectToJson(dataContent)));
            }
        }
    }
    /**
     *  客户端连接到服务端时触发
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel currentChannel = ctx.channel();
        String currentChannelId = currentChannel.id().asLongText();

        System.out.println("客户端连接,channel对应的长id：" + currentChannelId);
        // 将当前通道加入到通道组进行管理
        clients.add(currentChannel);
    }

    /**
     *  客户端断开连接时触发
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel currentChannel = ctx.channel();
        String currentChannelId = currentChannel.id().asLongText();
        System.out.println("客户端断开连接,channel对应的长id：" + currentChannelId);

        // 移除多余的会话
        String userId = UserChannelSession.getUserIdByChannelId(currentChannelId);
        UserChannelSession.unbindMultiSession(userId,currentChannelId);
        UserChannelSession.unbindUser(currentChannelId);
        clients.remove(currentChannel);

        // 从redis中获取发送方所链接的netty结点
        Jedis jedis = JedisPoolUtils.getJedis();
        NettyServerNode nettyServerNode = JsonUtils.jsonToPojo(jedis.get(userId), NettyServerNode.class);
        // 断开连接时，对netty结点对象的在线人数减1
        ZookeeperRegister.decrementOnlineCounts(nettyServerNode);
    }

    /**
     *  异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel currentChannel = ctx.channel();
        String currentChannelId = currentChannel.id().asLongText();

        System.out.println("发生捕获异常,channel对应的长id：" + currentChannelId);

        // 关闭通道
        ctx.channel().close();

        // 移除当前通道
        clients.remove(currentChannel);

        String userId = UserChannelSession.getUserIdByChannelId(currentChannelId);
        UserChannelSession.unbindMultiSession(userId,currentChannelId);
        UserChannelSession.unbindUser(currentChannelId);
    }
}
